package org.example.flink.source;

import cn.hutool.core.date.DateUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.connector.pulsar.source.PulsarSource;
import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.pulsar.client.api.SubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE;

public class Flink14PS {

    private static final Logger logger = LoggerFactory.getLogger(Flink14PS.class);

    public static void main(String[] args) throws Exception {
        String serviceUrl = "pulsar://127.0.0.1:6650";
        String adminUrl = "http://127.0.0.1:8080";

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.setParallelism(1);

        PulsarSource<String> pulsarSource = PulsarSource.builder()
                .setServiceUrl(serviceUrl)
                .setAdminUrl(adminUrl)
                .setStartCursor(StartCursor.earliest())
                .setTopics("persistent://public/default/delay10")
                .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
                .setSubscriptionName("my2")
                .setConfig(PulsarSourceOptions.PULSAR_CONSUMER_NAME,"a1")
                .setSubscriptionType(SubscriptionType.Exclusive)
                .setUnboundedStopCursor(StopCursor.never())
                .setConfig(PulsarSourceOptions.PULSAR_SUBSCRIPTION_MODE, SubscriptionMode.Durable)
//                .setConfig(PulsarSourceOptions.PULSAR_RETRY_ENABLE,true)
//                .setConfig(PulsarOptions.PULSAR_ENABLE_TRANSACTION,false)
                .setConfig(PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE,true)
//                .setConfig(PulsarSourceOptions.PULSAR_MAX_FETCH_TIME,1L)
                .setConfig(PulsarSourceOptions.PULSAR_MAX_FETCH_RECORDS,1)
                .setConfig(PulsarSourceOptions.PULSAR_AUTO_COMMIT_CURSOR_INTERVAL,1000L)
                .build();

        DataStreamSource<String> pulsar_source = env.fromSource(pulsarSource, WatermarkStrategy.noWatermarks(), "Pulsar Source").setParallelism(1);

        pulsar_source.addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
//                logger.info("Msg:{} now={}",value, DateUtil.now());
                System.out.println("Msg:"+value+" now:"+DateUtil.now());
            }
        });

        final FileSink<String> fileSink = FileSink
                .forRowFormat(new Path("file:///home/zhangjg/tmpData/t1"), new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(TimeUnit.SECONDS.toMillis(10))
                                .withInactivityInterval(TimeUnit.SECONDS.toMillis(3))
                                .withMaxPartSize(1024 * 1024 * 1024)
                                .build())
                .build();

        pulsar_source.sinkTo(fileSink);

        env.execute("Flink14PS");

    }

}
